package com.test45_JUC;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * ForkJoinPool.ManagedBlocker，分支任务阻塞管理
 */
public class Demo31_ManagedBlocker {

    public static void main(String[] args) throws Exception {
        SumTask4 task = new SumTask4(1, 100);
        ForkJoinPool pool = new ForkJoinPool(2);  // 开启分支任务池
        Future<Integer> future = pool.submit(task);  // 执行分支任务
        System.out.println("分支任务计算结果：" + future.get());
    }
}

/**
 * 实现数据累加的计算
 */
class SumTask4 extends RecursiveTask<Integer> {
    private static final int THRESHOLD = 25;    // 分支阈值
    private int start;  // 开始计算数值
    private int end;    // 结束计算数值
    private Lock lock = new ReentrantLock();

    public SumTask4(int start, int end) {
        this.start = start;
        this.end = end;
    }

    // 完成计算的处理
    @Override
    protected Integer compute() {
        // 所有的子分支的处理，以及所有相关分支的合并处理都在此方法之中完成
        boolean isFork = (end - start) > THRESHOLD;    // 是否需要进行分支
        if (isFork) {   // 需要开启分支
            int middle = (start + end) / 2;
            SumTask leftTask = new SumTask(this.start, middle);
            SumTask rightTask = new SumTask(middle + 1, this.end);
            leftTask.fork();    // 开启左分支
            rightTask.fork();   // 开启右分支
            return leftTask.join() + rightTask.join();   // 分之结果合并
        } else {    // 计算子分支
            SumHandleManagedBlocker blocker = new SumHandleManagedBlocker(this.start, this.end, this.lock);
            try {
                ForkJoinPool.managedBlock(blocker);  // 加入阻塞管理
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return blocker.result;  // 返回计算的结果
        }
    }

    /**
     * 自定义线程管理
     */
    static class SumHandleManagedBlocker implements  ForkJoinPool.ManagedBlocker {
        private Integer result;
        private int start;
        private int end;
        private Lock lock;

        public SumHandleManagedBlocker(int start, int end, Lock lock) {
            this.start = start;
            this.end = end;
            this.lock = lock;
        }

        // 处理延迟任务
        @Override
        public boolean block() throws InterruptedException {
            int sum = 0;
            this.lock.lock();
            try {
                for (int x = start; x <= end ; x++) {
                    TimeUnit.MILLISECONDS.sleep(100);
                    sum += x;
                }
            } finally {
                this.result = sum;
                this.lock.unlock();
            }
            System.out.printf("【%s】处理数据累加业务，start = %d、end = %d、sum = %d%n",
                    Thread.currentThread().getName(), this.start, this.end, sum);
            return result != null;  // 结束标记
        }

        // 补偿的判断，返回false会创建补偿线程
        @Override
        public boolean isReleasable() {
            return result != null;  // 阻塞解除判断
        }
    }
}
